package com.dyna.cp.binlog;

import cn.hutool.core.collection.CollUtil;
import com.dyna.constants.StringPool;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import com.github.shyiko.mysql.binlog.event.*;
import com.github.shyiko.mysql.binlog.event.deserialization.EventDeserializer;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.Serializable;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Component
public class BinaryLogClientRunner implements CommandLineRunner {

    private final BinLogConstants binLogConstants;
    private static String host;
    private static int port;
    private static String username;
    private static String passwd;
    private static Integer serverId;
    private static List<String> tableList;
    private static Boolean startFlag;
    private final BinLogService binLogService;

    @Autowired
    public BinaryLogClientRunner(BinLogConstants binLogConstants,
                                 BinLogService binLogService
                                 ) {
        this.binLogConstants = binLogConstants;
        BinaryLogClientRunner.host = binLogConstants.getHost();
        BinaryLogClientRunner.port = binLogConstants.getPort();
        BinaryLogClientRunner.username = binLogConstants.getUsername();
        BinaryLogClientRunner.passwd = binLogConstants.getPasswd();
        BinaryLogClientRunner.serverId = binLogConstants.getServerId();
        BinaryLogClientRunner.tableList = binLogConstants.getTables();
        BinaryLogClientRunner.startFlag = binLogConstants.getStartFlag();
        this.binLogService = binLogService;
    }

    @Data
    @Builder
    @AllArgsConstructor
    public static class TableData {
        private String database;
        private String table;
        private String databaseTable;
    }

    private static final Map<String, TableData> map = new ConcurrentHashMap<>();

    @Override
    public void run(String... args) throws Exception {
        if (!startFlag) {
            return;
        }
        // 数据库下的表字段
        Map<String, Map<Integer, String>> metadata = BingLogMetadata.database(binLogConstants);
        /**
         * BinaryLogClient 类是 MySQL 提供的一个 Java 客户端，用于监听 MySQL 数据库的二进制日志（Binary Log），并实现实时的数据同步。
         * BinaryLogClient 库底层使用了 MySQL 原生协议和通信机制，能够准确地捕获到 MySQL 数据库中的变化事件，
         * 并将这些事件以异步回调的方式传递给应用程序进行处理
         */
        BinaryLogClient client = new BinaryLogClient(host, port, username, passwd);
        /**
         * EventDeserializer 是一个接口，主要用于将 Event 数据从序列化格式反序列化为对象格式
         */
        // 数据序列化为java 对象
        EventDeserializer eventDeserializer = new EventDeserializer();
        eventDeserializer.setCompatibilityMode(
                // 此模式下，时间戳表示为自 Unix 纪元（1970 年 1 月 1 日 UTC）以来的毫秒数。这与其他兼容模式不同，后者可能使用不同的格式来表示时间戳
                EventDeserializer.CompatibilityMode.DATE_AND_TIME_AS_LONG,
                // 在此模式下，通过使用字节数组来表示字符和二进制数据，同时保留其编码和格式信息。这使得在读取时更容易处理这些数据，并将它们重新还原为原始格式
                EventDeserializer.CompatibilityMode.CHAR_AND_BINARY_AS_BYTE_ARRAY
        );
        client.setEventDeserializer(eventDeserializer);
        // 在BinaryLogClient中，serverId属性是一个客户端与MySQL实例之间的唯一标识符
        // 如果有多个客户端同时订阅同一个MySQL实例上的二进制日志，那么每个客户端的serverId应该是唯一的，以避免冲突
        client.setServerId(serverId);

        // 监听db 库中的数据变化
        client.registerEventListener(event -> {
            EventHeader header = event.getHeader();
            EventType eventType = header.getEventType();
            TableData tableData;

            if (eventType == EventType.TABLE_MAP) {
                // 构建SQL事件表信息
                buildEventTable(event);
            }else if
                    // 新增SQL
            (EventType.isWrite(eventType)) {
                WriteRowsEventData data = event.getData();
                if (null != (tableData = isListener(data.getTableId(), tableList))) {
                    log.info("--------------MySQL 新增数据--------------");
                    List<Serializable[]> rows = data.getRows();
                    binLogService.handleWrite(rows, tableData, metadata);
                }
            } else if
                    // 更新SQL
            (EventType.isUpdate(eventType)) {
                UpdateRowsEventData data = event.getData();
                if (null != (tableData = isListener(data.getTableId(), tableList))) {
                    log.info("--------------MySQL 修改数据--------------");
                    List<Map.Entry<Serializable[], Serializable[]>> rows = data.getRows();
                    binLogService.handleUpdate(rows, tableData, metadata);
                }
            } else if
                    // 删除SQL
            (EventType.isDelete(eventType)) {
                DeleteRowsEventData data = event.getData();
                if (null != (tableData = isListener(data.getTableId(), tableList))) {
                    log.info("--------------MySQL 删除数据--------------");
                    List<Serializable[]> rows = data.getRows();
                    binLogService.handleDelete(rows, tableData, metadata);
                }
            }
        });
        try {
            client.connect();
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * TableMapEventData，通过它可以获取操作的数据库名称、表名称以及表的id。之所以我们要监听这个事件，
     * 是因为之后监听的实际操作中返回数据中包含了表的id，而没有表名等信息，
     * 所以如果我们想知道具体的操作是在哪一张表的话，就要先维护一个id与表的对应关系
     */
    private static void buildEventTable(com.github.shyiko.mysql.binlog.event.Event event) {
        TableMapEventData eventData = event.getData();
        long tableId = eventData.getTableId();
        String database = eventData.getDatabase();//库名
        String table = eventData.getTable();//表名称
        StringBuilder builder = new StringBuilder();
        builder.append(database).append(StringPool.DOT).append(table);
        if (tableList.contains(builder.toString())) {
            map.put(String.valueOf(tableId), TableData.builder().database(database).table(table).build());
        }
    }

    /**
     * 当前监听事件表是否在监听列表中
     * @param tableId 当前监听event中的表id
     * @param tableList 监听表id
     * @return TableData
     */
    private TableData isListener(long tableId, List<String> tableList) {
        TableData tableData = map.get(String.valueOf(tableId));
        if (null == tableData || CollUtil.isEmpty(tableList))
            return null;
        String database = tableData.getDatabase();
        String table = tableData.getTable();
        StringBuilder builder = new StringBuilder();
        builder.append(database).append(StringPool.DOT).append(table);
        if (tableList.contains(builder.toString())) {
            tableData.setDatabaseTable(builder.toString());
            return tableData;
        }
        return null;
    }

}
